package service

import (
	"fmt"
	"gitee.com/crack007/goose/config"
	"gitee.com/crack007/goose/log"
	"github.com/crack007/go-im/src/common/constant"
	"github.com/streadway/amqp"
)

type Queue struct {
	channel      *amqp.Channel
	connection   *amqp.Connection
	routingKey   string
	exchange     string
	queueName    string
	exchangeType constant.ExchangeType
}

func (p *Queue) Init() (err error) {
	url := fmt.Sprintf("amqp://%s:%s@%s:%d%s",
		config.GetString("rabbitmq.user", "admin"),
		config.GetString("rabbitmq.password", "admin"),
		config.GetString("rabbitmq.host", "127.0.0.1"),
		config.GetInt("rabbitmq.port", 5379),
		config.GetString("rabbitmq.vhost", "/"),
	)
	log.GetLogger().Debug("Connect rabbitmq-> %s", url)
	p.connection, err = amqp.Dial(url)
	if err != nil {
		return err
	}
	p.channel, err = p.connection.Channel()
	return err
}

func (p *Queue) ExchangeDeclare() (err error) {
	err = p.channel.ExchangeDeclare(p.exchange, string(p.exchangeType), true, false, false, true, nil)
	return err
}
func (p *Queue) Close() {
	log.GetLogger().Trace("Queue Close")
	p.channel.Close()
	p.connection.Close()
}

func (p *Queue) Publish(data string) error {
	err := p.channel.Publish(p.exchange, p.routingKey, true, false, amqp.Publishing{
		ContentType: "text/plain",
		Body:        []byte(data),
	})
	if err != nil {
		return err
	}
	return nil
}
func (p *Queue) QueueDeclare() (err error) {
	_, err = p.channel.QueueDeclare(p.queueName, true, false, false, false, nil)
	err = p.channel.QueueBind(p.queueName, p.routingKey, p.exchange, true, nil)
	return err
}

func (p *Queue) Consumer(callback func(amqp.Delivery), consumer string) {
	msg, err := p.channel.Consume(p.queueName, consumer, false, false, false, false, nil)
	if err != nil {

		return
	}
	for delivery := range msg {
		callback(delivery)
	}
}
func (p *Queue) Cancel(consumer string) {
	p.channel.Cancel(consumer, false)
}
func NewQueue(routingKey string, exchange string, queueName string, exchangeType constant.ExchangeType) *Queue {
	return &Queue{
		routingKey:   routingKey,
		exchange:     exchange,
		queueName:    queueName,
		exchangeType: exchangeType,
	}
}

func NewMessageQueue() *Queue {
	return NewQueue("message", "messageExchange", "imQueue", constant.ExchangeTypeDirect)
}

func NewWebsocketQueue(routingKey string) *Queue {
	return NewQueue(routingKey, "wsExchange", "wsQueue", constant.ExchangeTypeDirect)
}
